package com.sui.bigdata.flink.sql.side;


import com.sui.bigdata.flink.sql.core.Main;
import org.apache.flink.calcite.shaded.com.google.common.base.Charsets;
import org.apache.flink.calcite.shaded.com.google.common.collect.Lists;
import org.junit.Test;
import com.sui.bigdata.flink.sql.function.scalar.CurrentTime;

import java.net.URLDecoder;
import java.net.URLEncoder;
import java.util.Arrays;
import java.util.List;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * @author YongChen
 * @date 2019/8/29 12:21
 * @description
 * @email yong_chen@sui.com
 */
public class demo {


    //{"suid":"9527","click":"1","time1":"1583994915000"}
    //{"suid":"9527","click":"1","time1":"1583994925000"}

    //{"suid":"9527","click":"2","time1":"1583994935000"}
    //{"suid":"9527","click":"2","time1":"1583994955000"}

    @Test
    public void testTopN() throws Exception {
        String sql =" CREATE TABLE MyTable(\n" +
                "    type varchar as type,\n" +
                "    suid varchar as suid,\n" +
                "    click int as click \n" +
                " )WITH(\n" +
                "    type ='kafka',\n" +
                "    bootstrapServers ='10.201.7.187:9093,10.201.7.188:9093,10.201.7.189:9093,10.201.7.190:9093',\n" +
                "    offsetReset ='earliest',\n" +
                "    groupId = 'flinkwindow_test1',\n" +
                "    topic ='flinkwindow_test',\n" +
                "    parallelism ='1'\n" +
                " );\n" +
                " SELECT type,suid,click ,row_num " +
                "  FROM (  " +
                "     SELECT type,suid,click,  " +
                "         ROW_NUMBER() OVER (PARTITION BY type ORDER BY click DESC) as row_num \n" +
                "     FROM MyTable) " +
                "  WHERE row_num <= 2 ;";


        test1(sql);
    }

    @Test
    public void testWindow() throws Exception {
        String sql =" CREATE TABLE MyTable(\n" +
                "    suid varchar as suid,\n" +
                "    click int as click,\n" +
                "    time1 bigint as time1, "+
                " WATERMARK  FOR time1 AS withOffset(time1 ,5000)\n" +
                " )WITH(\n" +
                "    type ='kafka',\n" +
                "    bootstrapServers ='10.201.7.187:9093,10.201.7.188:9093,10.201.7.189:9093,10.201.7.190:9093',\n" +
                "    offsetReset ='latest',\n" +
                "    groupId = 'flinkwindow_test3',\n" +
                "    topic ='flinkwindow_test',\n" +
                "    parallelism ='1'\n" +
                " );\n" +
                " select \n" +
                "   suid,\n" +
//                "   cast(TUMBLE_START(time1, INTERVAL '2' SECOND) as varchar),\n" +
//                "   TUMBLE_END(time1, INTERVAL '2' SECOND),\n" +
                "   sum(click) \n" +
                "from \n" +
                "   MyTable \n" +
                "GROUP BY TUMBLE(time1,INTERVAL '2' SECOND) , suid;";


        test1(sql);
    }

    @Test
    public void testWindow1() throws Exception {
        String sql =" CREATE TABLE MyTable(\n" +
                "    suid varchar as suid,\n" +
                "    click int as click\n" +
                " )WITH(\n" +
                "    type ='kafka',\n" +
                "    bootstrapServers ='10.201.7.115:9092,10.201.7.120:9092,10.201.7.121:9092',\n" +
                "    offsetReset ='latest',\n" +
                "    groupId = 'flinkwindow_test1',\n" +
                "    topic ='flinkwindow_test',\n" +
                "    parallelism ='1'\n" +
                " );\n" +
                " select \n" +
                "   suid,\n" +
//                "   TUMBLE_START(time1, INTERVAL '10' SECOND),\n" +
//                "   TUMBLE_END(time1, INTERVAL '10' SECOND),\n" +
                "   sum(click) \n" +
                "from \n" +
                "   MyTable \n" +
                "GROUP BY TUMBLE(proctime,INTERVAL '10' SECOND) , suid;";


        test3(sql);
    }

    @Test
    public void testCountDis() throws Exception {
        String sql =" CREATE TABLE MyTable(\n" +
                "    suid varchar as suid,\n" +
                "    click int as click\n" +
                " )WITH(\n" +
                "    type ='kafka',\n" +
                "    bootstrapServers ='10.201.7.115:9092,10.201.7.120:9092,10.201.7.121:9092',\n" +
                "    offsetReset ='latest',\n" +
                "    groupId = 'flinkwindow_test1',\n" +
                "    topic ='flinkwindow_test',\n" +
                "    parallelism ='1'\n" +
                " );\n" +
                " select \n" +
                " distinct suid \n" +
                " from \n" +
                "   MyTable \n" +
                ";";


        test3(sql);
    }

    @Test
    public void testSinkKafka() throws Exception {
        String sql =  " "+
                " CREATE TABLE MyTable(\n" +
                "    suid varchar,\n" +
                "    click int\n" +
                " )WITH(\n" +
                "    type ='kafka',\n" +
                "    bootstrapServers ='10.201.7.115:9092,10.201.7.120:9092,10.201.7.121:9092',\n" +
                "    offsetReset ='latest',\n" +
                "    groupId = 'flinkwindow_test1',\n" +
                "    topic ='flinkwindow_test',\n" +
                "    parallelism ='1'\n" +
                " );\n" +

                " CREATE TABLE kafkaSink(\n" +
                "    suid varchar,\n" +
                "    click int\n" +
                " )WITH(\n" +
                "    type ='kafka',\n" +
                "    bootstrapServers ='10.201.7.187:9093,10.201.7.188:9093,10.201.7.189:9093,10.201.7.190:9093',\n" +
                "    topic ='datateam.riskctrl.rtc_admin_show',\n" +
                "    semantic ='at_least_once',\n" +
                "    parallelism ='1'\n" +
                " );\n" +

                " insert into kafkaSink "+
                "select  \n" +
                "   suid,\n" +
                "   click \n" +
                "from \n" +
                "   MyTable; \n";




        test2(sql);
    }

    @Test
    public void testsum() throws Exception {
        String sql =  " "+
                " CREATE TABLE MyTable(\n" +
                "    suid varchar AS suid,\n" +
                "    click int AS click\n" +
                " )WITH(\n" +
                "    type ='kafka',\n" +
                "    bootstrapServers ='10.201.7.187:9093,10.201.7.188:9093,10.201.7.189:9093,10.201.7.190:9093',\n" +
                "    offsetReset ='latest',\n" +
                "    groupId = 'SSSS' ,\n" +
                "    topic ='flinkwindow_test',\n" +
                "    parallelism ='1'\n" +
                " );\n" +

                "select" +
                "   suid,  \n" +
                "   max(click) \n" +
                "from \n" +
                "   MyTable group by suid ; \n";




        test2(sql);
    }

    @Test
    public void testMessage() throws Exception {
        String sql =  " "+
                " CREATE TABLE MyTable(\n" +
                "    busId varchar,\n" +
                "    empty_rate varchar\n" +
                " )WITH(\n" +
                "    type ='kafka',\n" +
                "    bootstrapServers ='10.201.7.115:9092,10.201.7.120:9092,10.201.7.121:9092',\n" +
                "    offsetReset ='latest',\n" +
                "    topic ='flinkwindow_test',\n" +
                "    parallelism ='1'\n" +
                " );\n" +

                " CREATE TABLE alarmMsg(   \n" +
                "    project varchar,\n" +
                "    busId varchar,\n" +
                "    des  varchar " +
                " )WITH(  \n" +
                "    type ='message',   \n" +
                "    phone ='15170603850',\n" +
                "    template ='[{project}] {busId}发生异常:{des}.'\n" +
                " );"+

                " insert into alarmMsg "+
                "select  \n" +
                "  '名单服务' as project ," +
                "   busId,\n" +
                "   concat('一小时内空值超过',empty_rate,'%') as des \n" +
                "from \n" +
                "   MyTable; \n";




        test2(sql);
    }

    @Test
    public void testSinkKudu() throws Exception {
        String sql =
                "CREATE TABLE MyTable(\n" +
                        "    suid varchar,\n" +
                        "    click int \n" +
                        " )WITH(\n" +
                        "    type ='kafka',\n" +
                        "    bootstrapServers ='10.201.7.115:9092,10.201.7.120:9092,10.201.7.121:9092',\n" +
                        "    offsetReset ='latest',\n" +
                        "    topic ='flinkwindow_test',\n" +
                        "    parallelism ='1'\n" +
                        " );\n" +
                        "CREATE TABLE kudusink (\n" +
                        " suid varchar ,\n " +
                        " click int  \n " +
                        " ) WITH ( \n" +
                        " type ='kudu',\n" +
                        "    kuduMasters ='10.201.7.125:7051,10.201.7.123:7051,10.201.7.124:7051',\n" +
                        "    tableName ='impala::dw.flink_test',\n" +
                        "    writeMode='insert',\n" +
                        "    parallelism ='1');"+
                        " insert into kudusink " +
                        " select suid,click from MyTable ;";


        test2(sql);
    }

    @Test
    public void testJoinKudu() throws Exception {

        //bj1-203-centos11:7051,bj1-203-centos12:7051,bj1-203-centos13:7051
        String sql =
                "CREATE TABLE MyTable(\n" +
                        "    suid varchar,\n" +
                        "    name varchar \n" +
                        " )WITH(\n" +
                        "    type ='kafka',\n" +
                        "    bootstrapServers ='10.201.7.187:9093,10.201.7.188:9093,10.201.7.189:9093,10.201.7.190:9093',\n" +
                        "    offsetReset ='latest',\n" +
                        "    groupId = 'SSSS' ,\n" +
                        "    topic ='flinkwindow_test',\n" +
                        "    parallelism ='1'\n" +
                        " );\n" +
                        "create table kuduSide(\n" +
                        "    suid varchar,\n" +
                        "    click int,\n" +
                        "    PRIMARY KEY(suid),\n" +
                        "    PERIOD FOR SYSTEM_TIME\n" +
                        " )WITH(\n" +
                        "    type='kudu',\n" +
                        "    kuduMasters ='10.201.7.125:7051,10.201.7.123:7051,10.201.7.124:7051',\n" +
                        "    tableName ='impala::dw.flink_test',\n" +
//                        "    workerCount='1',\n" +
                        "    defaultOperationTimeoutMs='600000',\n" +
                        "    defaultSocketReadTimeoutMs='6000000',\n" +
                        "    batchSizeBytes='100000000',\n" +
                        "    limitNum='1000',\n" +
                        "    isFaultTolerant='false',\n" +
                        "    partitionedJoin='false'\n" +
                        " );"+


                        " select t1.suid,t1.name,t2.click " +
                        "  from MyTable t1 join kuduSide t2 " +
                        "  on t1.suid = t2.suid and t1.name ='jack' and t2.click ='10' ;";




        test2(sql);
    }

    @Test
    public void twoStreamJoin() throws Exception {

        //bj1-203-centos11:7051,bj1-203-centos12:7051,bj1-203-centos13:7051
        String sql =
                "CREATE TABLE stream1(\n" +
                        "    suid varchar,\n" +
                        "    name varchar \n" +
                        " )WITH(\n" +
                        "    type ='kafka',\n" +
                        "    bootstrapServers ='10.201.7.187:9093,10.201.7.188:9093,10.201.7.189:9093,10.201.7.190:9093',\n" +
                        "    offsetReset ='latest',\n" +
                        "    groupId = 'SSSS' ,\n" +
                        "    topic ='flinkwindow_test',\n" +
                        "    parallelism ='1'\n" +
                        " );\n" +
                        "  CREATE TABLE stream2(\n" +
                        "    suid varchar,\n" +
                        "    sex varchar \n" +
                        " )WITH(\n" +
                        "    type ='kafka',\n" +
                        "    bootstrapServers ='10.201.7.187:9093,10.201.7.188:9093,10.201.7.189:9093,10.201.7.190:9093',\n" +
                        "    offsetReset ='latest',\n" +
                        "    groupId = 'SSSS' ,\n" +
                        "    topic ='flinkwindow_test1',\n" +
                        "    parallelism ='1'\n" +
                        " );\n" +


                        "  SELECT\n" +
                        "  o.suid, o.name, r.sex \n" +
                        "FROM\n" +
                        "  stream1 AS o \n" +
                        "  JOIN stream2 FOR SYSTEM_TIME AS OF o.rowtime AS r\n" +
                        "  ON o.suid = r.suid  ;"

//
//        expectExceptionThrown(
//                "SELECT * FROM MyTable AS T JOIN temporalTest " +
//                        "FOR SYSTEM_TIME AS OF T.rowtime AS D ON T.a = D.id",
//                "Temporal table join currently only supports 'FOR SYSTEM_TIME AS OF' " +
//                        "left table's proctime field",
//                classOf[TableException])

//                        " select t1.suid,t1.name,t2.sex " +
//                        "  from stream1 t1  join stream2 t2 " +
//                        "   on t1.suid = t2.suid " +
//                        "   and t1.proctime BETWEEN t2.proctime - INTERVAL '30' SECOND AND t2.proctime + INTERVAL '30' SECOND ;"


                ;




        test2(sql);
    }

    @Test
    public void testTidb() throws Exception {
        String sql =
                "CREATE TABLE MyTable(\n" +
                "    suid varchar,\n" +
                "    click int \n" +
                " )WITH(\n" +
                "    type ='kafka',\n" +
                "    bootstrapServers ='10.201.7.115:9092,10.201.7.120:9092,10.201.7.121:9092',\n" +
                "    offsetReset ='latest',\n" +
                "    topic ='flinkwindow_test',\n" +
                "    parallelism ='1'\n" +
                " );\n" +
                "CREATE TABLE sideTable (\n" +
                " suid varchar ,\n " +
                " name varchar , \n " +
                " PRIMARY KEY(suid) , \n " +
                " PERIOD FOR SYSTEM_TIME \n " +
                " ) WITH ( \n" +
                " type='mysql', " +
                " url='jdbc:mysql://tidb1.testfeideedba.com:4000/dt?useUnicode=true&amp;characterEncoding=utf8;useSSL=true',"+
                " userName='DSJ_user',"+
                " password='606cde3507b026H,dHb_77e67H_e',"+
                " tableName='flink_join_tidb_test', " +
                " maxPoolSize = '100' , checkOutTimeout = '100000' );\n" +
                " select \n" +
                "   a.suid,\n" +
                "   cast(a.click as varchar),\n" +
                "   b.name " +
                "from  \n" +
                "  MyTable a  \n" +
                "  JOIN sideTable b  \n" +
                "  on a.suid=b.suid and b.name <>'Hanmeimei1' and a.click < 2 ;";


        test2(sql);
    }

    @Test
    public void testJoinPhoenix() throws Exception {
        String sql =
                "CREATE TABLE MyTable(\n" +
                        "    ID varchar ,\n" +
                        "    COL1 varchar ,\n" +
                        "    COL2 varchar \n" +
                        " )WITH(\n" +
                        "    type ='kafka',\n" +
                        "    bootstrapServers ='10.201.7.187:9093,10.201.7.188:9093,10.201.7.189:9093,10.201.7.190:9093',\n" +
                        "    offsetReset ='latest',\n" +
                        "    topic ='flinkwindow_test',\n" +
                        "    groupId = 'SSS',\n" +
                        "    parallelism ='1'\n" +
                        " );\n" +
                        "CREATE TABLE sideTable (\n" +
//                        " COL1 varchar ,\n " +
//                        " COL2 varchar , \n " +
//                        " PRIMARY KEY(ID) , \n " +

//                        " ID varchar ,\n " +
//                        " COL2 varchar , \n " +
//                        " PRIMARY KEY(COL1) , \n " +


                        " COL1 varchar ,\n " +
                        " ID varchar , \n " +
                        " PRIMARY KEY(COL2) , \n " +
                        " PERIOD FOR SYSTEM_TIME \n " +
                        " ) WITH ( \n" +
                        " type='mysql', " +
                        " url='jdbc:phoenix:10.201.7.125,10.201.7.126,10.201.7.127:2181',"+
//                        " userName='DSJ_user',"+
//                        " password='606cde3507b026H,dHb_77e67H_e',"+
                        " tableName='cy_phoenix_test', " +
                        " maxPoolSize = '100' , checkOutTimeout = '100000' );\n" +
                        " select \n" +
//                        "   a.ID,\n" +
//                        "   b.COL1 ,\n" +
//                        "   b.COL2 " +
//
//                        "   a.COL1,\n" +
//                        "   b.ID ,\n" +
//                        "   b.COL2 " +
//
//
                        "   a.COL2,\n" +
                        "   b.COL1 ,\n" +
                        "   b.ID " +
                        "from  \n" +
                        "  MyTable a  \n" +
                        "  JOIN sideTable b  \n" +
//                        "  on a.ID=b.ID ;";
//                        "  on a.COL1=b.COL1 ;";
                        "  on a.COL2=b.COL2 ;";



        test2(sql);
    }


    @Test
    public void testSinkPhoenix() throws Exception {
        String sql =
                "CREATE TABLE MyTable(\n" +
                        "   ID varchar,\n" +
                        "   COL1 varchar ,\n" +
                        "   COL2  varchar " +
                        " )WITH(\n" +
                        "    type ='kafka',\n" +
                        "    bootstrapServers ='10.201.7.187:9093,10.201.7.188:9093,10.201.7.189:9093,10.201.7.190:9093',\n" +
                        "    offsetReset ='latest',\n" +
                        "    topic ='flinkmysql_test',\n" +
                        "    groupId = 'SSS',\n" +
                        "    parallelism ='1'\n" +
                        " );\n" +
                        "CREATE TABLE MyResult (\n" +
                        "   ID    varchar,\n" +
                        "   COL1  varchar ,\n" +
                        "   COL2  varchar " +
                        " ) WITH ( \n" +
                        " type='phoenix', " +
                        " url='jdbc:phoenix:10.201.7.125,10.201.7.126,10.201.7.127:2181',"+
                        " tableName='cy_phoenix_test'," +
                        " batchSize ='500'," +
                        " batchWaitInterval = '5000',"+
                        " operation ='upsert'," +
                        " parallelism ='1' );\n" +

                        "insert into MyResult \n" +
                        "select \n" +
                        "   ID,\n" +
                        "   COL1 , " +
                        "   COL2  " +
                        "from MyTable;";


        test1(sql);
    }

    @Test
    public void testWindowGroup() throws Exception {
        String sql =" CREATE TABLE MyTable(\n" +
                "    suid varchar,\n" +
                "    click int ,\n" +
                "    time1 timestamp , "+
                " WATERMARK  FOR time1 AS withOffset(time1 ,10)\n" +
                " )WITH(\n" +
                "    type ='kafka',\n" +
                "    bootstrapServers ='10.201.7.115:9092,10.201.7.120:9092,10.201.7.121:9092',\n" +
                "    offsetReset ='latest',\n" +
                "    groupId = 'flinkwindow_test1',\n" +
                "    topic ='flinkwindow_test',\n" +
                "    parallelism ='1'\n" +
                " );\n" +
                " select \n" +
                "   suid,\n" +
                "   TUMBLE_START(time1, INTERVAL '10' SECOND),\n" +
                "   TUMBLE_END(time1, INTERVAL '10' SECOND),\n" +
                "   sum(click) \n" +
                "from \n" +
                "   MyTable \n" +
                "GROUP BY TUMBLE(time1,INTERVAL '10' SECOND) , suid;";


        test1(sql);
    }



    @Test
    public void testJoinHbase() throws Exception {

//        {"suid":"9527","fname":"u_ssj"}
        String sql =
                "CREATE TABLE MyTable(\n" +
                "    suid varchar ,\n" +
                "    fname varchar "+
//                "    time1 timestamp , "+
//                " WATERMARK  FOR time1 AS withOffset(time1 ,10)\n" +
                " )WITH(\n" +
                "    type ='kafka',\n" +
                "    bootstrapServers ='10.201.7.115:9092,10.201.7.120:9092,10.201.7.121:9092',\n" +
                "    offsetReset ='latest',\n" +
                "    groupId ='flinkwindow_test1',\n" +
                "    topic ='flinkwindow_test',\n" +
                "    parallelism ='1'\n" +
                " );\n" +
                "CREATE TABLE hbase_side (  \n" +
                "   info:name varchar as name, \n" +
                "   info:sex int as sex ,\n" +
                "   PRIMARY KEY (suid), \n" +
                "   PERIOD FOR SYSTEM_TIME ) WITH  \n" +
                "  (type = 'hbase', \n" +
                "   zookeeperQuorum = '10.201.7.186:2181,10.201.7.187:2181,10.201.7.188:2181', \n" +
                "   zookeeperParent = '/hbase', \n" +
                "   tableName = 'testflink'); \n" +

                "select \n" +
                "   a.suid as suid,\n" +
                "   b.name as name " +
//                "   a.time1 as time1 \n" +
                "from  MyTable a   join hbase_side b " +
                "   on a.suid=b.suid and b.name ='123' and b.sex>124 and a.fname='u_ssj' ;" ;




        test2(sql);
    }

    @Test
    public void testSinkMysql() throws Exception {
        String sql =
                "CREATE TABLE MyTable(\n" +
                        "    suid varchar ," +
                        "    name varchar \n" +
                        " )WITH(\n" +
                        "    type ='kafka',\n" +
                        "    bootstrapServers ='10.201.7.187:9093,10.201.7.188:9093,10.201.7.189:9093,10.201.7.190:9093',\n" +
                        "    offsetReset ='latest',\n" +
                        "    topic ='flinkmysql_test',\n" +
                        "    groupId ='flinkwindow_test',\n" +
                        "    parallelism ='1'\n" +
                        " );\n" +
                        "CREATE TABLE MyResult (\n" +
                        " suid varchar ,\n " +
                        " name varchar \n " +
                        " ) WITH ( \n" +
                        " type='mysql', " +
                        " url='jdbc:mysql://tidb1.testfeideedba.com:4000/dt?useUnicode=true&amp;characterEncoding=utf8;useSSL=true',"+
                        " userName='DSJ_user',"+
                        " password='606cde3507b026H,dHb_77e67H_e',"+
                        " tableName='flink_join_tidb_test'," +
                        " batchSize ='500'," +
                        " batchWaitInterval = '5000',"+
                        " operation ='insert' );"+
                        "insert into MyResult \n" +
                        "select \n" +
                        "   suid,\n" +
                        "   name " +
                        "from MyTable;";


        test1(sql);
    }

    @Test
    public void testaggregate() throws Exception {
        String sql =" create aggregate function PhoneHHI with com.feidee.flink.sql.function.aggregate.PhoneHHI ;"+
                "CREATE TABLE MyTable(\n" +
                "    suid varchar,\n" +
                "    phone varchar \n" +
                " )WITH(\n" +
                "    type ='kafka',\n" +
                "    bootstrapServers ='10.201.7.115:9092,10.201.7.120:9092,10.201.7.121:9092',\n" +
                "    offsetReset ='latest',\n" +
                "    topic ='flinkwindow_test',\n" +
                "    parallelism ='1'\n" +
                " );\n" +
                "CREATE TABLE MyResult (\n" +
                "  suid varchar,\n " +
                " tag bigint ,"+
                " tag1 bigint "+
                " ) WITH ( \n" +
                " type='console', " +
                " parallelism ='1');\n" +
                "insert into MyResult \n" +
                "select \n" +
                "  suid,\n" +
                "  PhoneHHI(phone,0,5), " +
                "  PhoneHHI(phone,-4)\n" +
                "from \n" +
                "   MyTable \n" +
                "GROUP BY SESSION(proctime,INTERVAL '5' SECOND), suid;";


        test2(sql);
    }
    @Test
    public void testOverWindow() throws Exception {
        String sql =" " +
                " create aggregate function PhoneHHI with com.feidee.flink.sql.function.aggregate.PhoneHHI ;"+
                "CREATE TABLE MyTable(\n" +
                "    suid varchar,\n" +
                "    click int \n" +
                " )WITH(\n" +
                "    type ='kafka',\n" +
                "    bootstrapServers ='10.201.7.115:9092,10.201.7.120:9092,10.201.7.121:9092',\n" +
                "    offsetReset ='latest',\n" +
                "    topic ='flinkoverwindow_test',\n" +
                "    parallelism ='1'\n" +
                " );\n" +
                "CREATE TABLE MyResult (\n" +
                "  suid varchar,\n " +
                "  sum int "+
                " ) WITH ( \n" +
                " type='console', " +
                " parallelism ='1');\n" +
                "insert into MyResult \n" +
                "SELECT suid,sum(click) OVER (\n" +
                "  PARTITION BY suid\n" +
                "  ORDER BY proctime\n" +
                "  ROWS BETWEEN 2 PRECEDING AND CURRENT ROW)\n" +
                "FROM MyTable; ";


        test2(sql);
    }

    @Test
    public void testcep() throws Exception {
        String sql =
                " CREATE TABLE Ticker(\n" +
                        " symbol    varchar, \n" +
                        " tax       varchar, \n" +
                        " price     double, \n" +
                        " rowtime        timestamp, \n" +
                        " WATERMARK  FOR rowtime AS withOffset(rowtime ,5)\n" +
                        " )WITH(\n" +
                        "    type ='kafka',\n" +
                        "    bootstrapServers ='10.201.7.115:9092,10.201.7.120:9092,10.201.7.121:9092',\n" +
                        "    offsetReset ='latest',\n" +
                        "    topic ='flink_cep1_test',\n" +
                        "    parallelism ='1'\n" +
                        " );\n" +
                        " CREATE TABLE MyResult (\n" +
                        "    symbol      varchar,\n" +
                        "    start_tstamp       timestamp,\n" +
                        "    end_tstamp       timestamp,\n" +
                        "    avgPrice       double \n" +
                        " ) WITH ( \n" +
                        " type='console'); \n"+
                        " insert into MyResult " +
                        " SELECT *\n" +
                        " FROM Ticker\n" +
                        "    MATCH_RECOGNIZE (\n" +
                        "        PARTITION BY symbol\n" +
                        "        ORDER BY rowtime\n" +
                        "        MEASURES\n" +
                        "            FIRST(A.rowtime) AS start_tstamp,\n" +
                        "            LAST(A.rowtime) AS end_tstamp,\n" +
                        "            AVG(A.price) AS avgPrice\n" +
                        "        ONE ROW PER MATCH\n" +
                        "        AFTER MATCH SKIP PAST LAST ROW\n" +
                        "        PATTERN (A+ B)\n" +
                        "        DEFINE\n" +
                        "            A AS AVG(A.price) < 15\n" +
                        "    ) MR;";


        test1(sql);
    }

    @Test
    public void testcep1() throws Exception {
        String sql =
                " CREATE TABLE bdl_suigo_business_apply (\n" +
                        "   rowData.customer_id varchar as customer_id,\n" +
                        "   rowData.business_type varchar as business_type\n" +
                        ") WITH (\n" +
                        "   type ='kafka',\n" +
                        "   bootstrapServers ='10.201.7.188:9093',\n" +
                        "   offsetReset ='earliest',\n" +
                        "   groupId='datateam-graph-201912261430',\n" +
                        "   topic ='rtsync.bdl_suigo_business_apply',\n" +
                        "   parallelism ='1'\n" +
                        ");\n" +
                        "\n" +
                        "CREATE TABLE test (\n" +
                        "   customer_id varchar,\n" +
                        "   business_type varchar\n" +
                        ") WITH (\n" +
                        "   type='console',\n" +
                        "   parallelism ='1'\n" +
                        ");\n" +
                        "\n" +
                        " insert into test\n" +
                        " select customer_id,business_type from bdl_suigo_business_apply;";


        test1(sql);
    }


    @Test
    public void testjson() throws Exception {
        String sql ="create scalar function GetYmd with com.sui.bigdata.flink.sql.function.scalar.GetYmd; "+
                " CREATE TABLE test_utc (\n" +
                        "   showtime timestamp as showtime\n" +
                        ") WITH (\n" +
                        "   type ='kafka',\n" +
                        "   bootstrapServers ='10.201.7.115:9092,10.201.7.120:9092,10.201.7.121:9092',\n" +
                        "   offsetReset ='latest',\n" +
                        "   groupid='datateam-realtime-report-onlinead-202001081507',\n" +
                        "   topic ='lc.bigdata.onlinead_clientshowlog',\n" +
                        "   parallelism='5' "+
                        ");  " +
                        "  select showtime,GetYmd(showtime) from test_utc;";


        test1(sql);
    }

    @Test
    public void testfun() throws Exception {
        String sql =
                " create scalar function UnixTimestamp  with com.feidee.flink.sql.function.scalar.UnixTimestamp ; "+
                " CREATE TABLE test (\n" +
                        "    suid varchar ," +
                        "    ts varchar \n" +
                        ") WITH (\n" +
                        "    type = 'kafka',\n" +
                        "    bootstrapServers = '10.201.7.115:9092',\n" +
                        "    offsetReset = 'latest',\n" +
                        "    topic = 'flink_cy_test',\n" +
                        "    groupId = 'dgs_applist',\n" +
                        "    parallelism = '1'\n" +
                        ");" +
                        " select  suid  from  test " +
                        " WHERE UnixTimestamp(cast(LOCALTIMESTAMP as varchar))-UnixTimestamp(ts) >0;";


        test1(sql);
    }

    @Test
    public void testMulJoin() throws Exception {
        String sql =
                "CREATE TABLE duotou_tidb_trick_1(\n" +
                        "    id_card varchar ,\n" +
                        "    phone varchar\n" +
                        " )WITH(\n" +
                        "    type ='kafka',\n" +
                        "    bootstrapServers ='10.201.7.187:9093,10.201.7.188:9093,10.201.7.189:9093,10.201.7.190:9093',\n" +
                        "    offsetReset ='latest',\n" +
                        "    groupid ='chenyong_duotou1',\n" +
                        "    topic ='datateam.riskctrl.duotou_tidb_trick',\n" +
                        "    parallelism ='1'\n" +
                        " ); \n" +
                        " \n" +
                        " \n" +
                        "CREATE TABLE duotou_id_udid_side_1 (\n" +
                        " id_card varchar ,\n" +
                        " udid varchar  ,\n" +
                        " PRIMARY KEY (id_card),\n" +
                        " PERIOD FOR SYSTEM_TIME\n" +
                        " ) WITH (\n" +
                        "   type='mysql',\n" +
                        "   url='jdbc:mysql://tidb1.testfeideedba.com:4000/dt?useUnicode=true&amp;characterEncoding=utf8;useSSL=true',\n" +
                        "   userName='DSJ_user',\n" +
                        "   password='606cde3507b026H,dHb_77e67H_e',\n" +
                        "    tableName='adl_bd_ls_duotou_id_phone_udid_side'); \n" +
                        "\t\n" +
                        "CREATE TABLE duotou_id_age_side_3(\n" +
                        "  tag:age bigint as age,\n" +
                        "  PRIMARY KEY(id_card),\n" +
                        "  PERIOD FOR SYSTEM_TIME\n" +
                        " ) WITH (\n" +
                        "   type ='hbase',\n" +
                        "   zookeeperQuorum ='10.201.7.113:2181,10.201.7.114:2181,10.201.7.115:2181',\n" +
                        "   zookeeperParent ='/hbase',\n" +
                        "   tableName ='id_age_tagging',\n" +
                        "   parallelism ='1'); \n" +
                        "\n" +
                        " CREATE TABLE duotou_id_name_side_4(\n" +
                        "    id_card varchar,\n" +
                        "    name varchar,\n" +
                        "    PRIMARY KEY(id_card),\n" +
                        "    PERIOD FOR SYSTEM_TIME\n" +
                        " )WITH(\n" +
                        "   type='mysql',\n" +
                        "   url='jdbc:phoenix:10.201.7.125,10.201.7.126,10.201.7.127:2181',\n" +
                        "   tableName='cy_phoenix_test_id4',\n" +
                        "   parallelism ='1' \n" +
                        " );\n" +
                        " \n" +
                        " \n" +
                        "CREATE TABLE id_info_result(\n" +
                        "  tag:id_md5  varchar,\n" +
                        "  tag:udid    varchar,\n" +
                        "  tag:age     bigint,\n" +
                        "  tag:name    varchar\n" +
                        " )WITH(\n" +
                        "  type='hbase',\n" +
                        "  zookeeperQuorum='10.201.7.113:2181,10.201.7.114:2181,10.201.7.115:2181',\n" +
                        "  zookeeperParent = '/hbase',\n" +
                        "  tableName='id_info_result',\n" +
                        "  rowkey='tag:id_md5'\n" +
                        ");  \n" +
                        "\n" +
                        "\n" +
//                        "INSERT INTO id_info_result \n" +
                        "SELECT\n" +
                        "   g1.id_card,\n" +
                        "   g2.udid, \n" +
                        "   g1.age,\n" +
                        "   g1.name\n" +
                        "FROM\n" +
                        "\t(\n" +
                        "\tSELECT\n" +
                        "\t\tk1.id_card,\n" +
                        "\t\tk1.age,\n" +
                        "\t\tk2.name \n" +
                        "\tFROM\n" +
                        "\t\t( SELECT t1.id_card, t2.age FROM duotou_tidb_trick_1 t1 JOIN duotou_id_age_side_3 t2 ON t1.id_card = t2.id_card ) k1\n" +
                        "\t\tJOIN duotou_id_name_side_4 k2 ON k1.id_card = k2.id_card \n" +
                        "\t) g1\n" +
                        "\tJOIN duotou_id_udid_side_1 g2 ON g1.id_card = g2.id_card \n" +
                        "\tAND g2.udid <> '' \n" +
                        "\tAND g1.name <> '';" +
                        "  SELECT id_card FROM duotou_tidb_trick_1 ;" +
                        " SELECT t1.id_card, t2.age FROM duotou_tidb_trick_1 t1 JOIN duotou_id_age_side_3 t2 ON t1.id_card = t2.id_card;" +
                        ";\n";


        test1(sql);
    }

    @Test
    public void testXXX() throws Exception {
        String str="sz-7-centos126.jobmanager.kafka_hbase.fullRestarts";
        System.out.println(str.split("\\.")[2]);
        Pattern p = Pattern.compile("^[-\\+]?[\\d]*$");

        String mobiles="123";
        if (mobiles != null){
            Matcher m = p.matcher(mobiles);
            System.out.println("1");
        }
        System.out.println(-1.0 / 0.0);
    }

    @Test
    public void testccc() throws Exception {


    }


    public void test2(String sql) throws Exception {

        List<String> paramList = Lists.newArrayList();
        paramList.add("-sql");
        String exeSql = URLEncoder.encode(sql, Charsets.UTF_8.name());
        paramList.add(exeSql);
        paramList.add("-name");
        paramList.add("xc");
        paramList.add("-mode");
        paramList.add("local");
        paramList.add("-port");
        paramList.add("8081");
        paramList.add("-confProp");
        String conf = "{\"time.characteristic\":\"ProcessingTime\",\"parallelism\":1,\"sql.select.sink\":\"console\" }";
//        String conf = "{\"time.characteristic\":\"EventTime\",\"parallelism\":1,\"sql.select.sink\":\"console\"," +
//                "\"mini.batch.allow.latency\":\"10000 ms\",\"mini.batch.size\":\"10\"}";
//                String conf = "{\"time.characteristic\":\"EventTime\",\"parallelism\":1,\"sql.select.sink\":\"console\"," +
//                "\"distinct.agg.split.bucket.num\":\"10\"}";
        String confEncode = URLEncoder.encode(conf, Charsets.UTF_8.name());
        paramList.add(confEncode);

        String[] params = new String[paramList.size()];
        paramList.toArray(params);
        Main.main(params);
    }


    public void test1(String sql) throws Exception {
        List<String> paramList = Lists.newArrayList();
        paramList.add("-sql");
        String exeSql = URLEncoder.encode(sql, Charsets.UTF_8.name());
        paramList.add(exeSql);
        paramList.add("-name");
        paramList.add("xc");
        paramList.add("-mode");
        paramList.add("local");
        paramList.add("-port");
        paramList.add("8083");
        paramList.add("-confProp");
        String conf = "{\"time.characteristic\":\"EventTime\",\"parallelism\":\"1\"}";
        String confEncode = URLEncoder.encode(conf, Charsets.UTF_8.name());
        paramList.add(confEncode);

        String[] params = new String[paramList.size()];
        paramList.toArray(params);
        Main.main(params);
    }

    public void test3(String sql) throws Exception {
        List<String> paramList = Lists.newArrayList();
        paramList.add("-sql");
        String exeSql = URLEncoder.encode(sql, Charsets.UTF_8.name());
        paramList.add(exeSql);
        paramList.add("-name");
        paramList.add("xc");
        paramList.add("-mode");
        paramList.add("local");
        paramList.add("-port");
        paramList.add("8081");
        paramList.add("-confProp");
        String conf = "{\"time.characteristic\":\"ProcessingTime\",\"parallelism\":1}";
        String confEncode = URLEncoder.encode(conf, Charsets.UTF_8.name());
        paramList.add(confEncode);

        String[] params = new String[paramList.size()];
        paramList.toArray(params);
        Main.main(params);
    }

//    public static void main(String[] args) throws Exception  {
//           P1Time p1Time =new P1Time();
//           Integer x=1;
//            Integer y=2;
//
//        }
    }

